/*
 * Copyright 2012 Google Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.google.dspl.validate.validator;

import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import com.google.dspl.model.Concept;
import com.google.dspl.model.DataType;
import com.google.dspl.model.Identifier;
import com.google.dspl.model.Slice;
import com.google.dspl.model.Slice.ConceptMapping;
import com.google.dspl.model.Slice.TableMapping;
import com.google.dspl.model.Table;
import com.google.dspl.model.Table.Column;
import com.google.dspl.validate.DatasetResolver;
import com.google.dspl.validate.ItemReference;
import com.google.dspl.validate.ValidationContext;
import com.google.dspl.validate.violation.SliceColumnNotFoundViolation;
import com.google.dspl.validate.violation.SliceConceptNotFoundViolation;
import com.google.dspl.validate.violation.SliceConflictingTypeViolation;
import com.google.dspl.validate.violation.SliceDuplicateConceptViolation;
import com.google.dspl.validate.violation.SliceNoIdViolation;
import com.google.dspl.validate.violation.SliceTableNotFoundViolation;
import com.google.dspl.validate.violation.Violation;

import java.util.List;
import java.util.Set;

/**
 * {@link Validator} for {@link Slice}s.
 *
 * @author James Grimm
 */
public class SliceValidator implements Validator<Slice> {

  @Override
  public List<Violation> validate(Slice slice, ValidationContext context) {
    ImmutableList.Builder<Violation> violations = ImmutableList.builder();

    validateId(slice, context, violations);
    validateTableMapping(slice, context, violations);

    validateConceptReferences(slice, context, violations);

    return violations.build();
  }

  private void validateConceptReferences(Slice slice, ValidationContext context,
      ImmutableList.Builder<Violation> violations) {

    Set<Identifier> uniqueReferencedConcepts = Sets.newHashSet();
    for (Identifier id : Iterables.concat(slice.getDimensions(), slice.getMetrics())) {
      if (uniqueReferencedConcepts.contains(id)) {
        violations.add(
            new SliceDuplicateConceptViolation(sliceReference(slice, context), slice.getId(), id));
      }
      uniqueReferencedConcepts.add(id);
      validateConceptReference(id, slice, context, violations);
    }
  }

  private void validateTableMapping(Slice slice, ValidationContext context,
      ImmutableList.Builder<Violation> violations) {
    TableMapping tableMapping = slice.getTable();
    Identifier tableId = new Identifier(context.getCurrentDataset().getDatasetId(),
        tableMapping.getTableId());
    validateTableReference(slice, tableId, context, violations);

    Iterable<ConceptMapping> allMappings =
        Iterables.concat(tableMapping.getDimensionMappings(), tableMapping.getMetricMappings());
    for (ConceptMapping conceptMapping : allMappings) {
      String columnId = conceptMapping.getToColumn();
      validateColumnReference(tableId, columnId, slice, context, violations);
      DatasetResolver resolver = context.getDatasetResolver();
      Column column = resolver.getColumn(tableId, columnId);
      DataType columnType = null;
      if (column != null) {
        columnType = column.getType();
      }
      Identifier conceptId = conceptMapping.getConcept();
      validateConceptReference(conceptId, slice, context, violations);

      Concept concept = context.getDatasetResolver().getConcept(conceptId);
      DataType conceptType = resolver.resolveConceptType(conceptId.getDatasetId(), concept);

      // Note that violations due to a null type will be handled on the column and concept
      // validators directly.
      if (columnType != null && conceptType != null) {
        if (!columnType.equals(conceptType)) {
          violations.add(new SliceConflictingTypeViolation(
              sliceReference(slice, context),
              slice.getId(),
              columnId,
              tableId.getObjectId(),
              conceptId,
              conceptType,
              columnType));
        }
      }
    }
  }

  private void validateConceptReference(Identifier conceptId, Slice slice,
      ValidationContext context, ImmutableList.Builder<Violation> violations) {
    DatasetResolver resolver = context.getDatasetResolver();
    Concept concept = resolver.getConcept(conceptId);
    if (concept == null) {
      violations.add(new SliceConceptNotFoundViolation(
          sliceReference(slice, context), slice.getId(), conceptId));
    }
  }

  private void validateColumnReference(Identifier tableId, String columnId, Slice slice,
      ValidationContext context, ImmutableList.Builder<Violation> violations) {
    DatasetResolver resolver = context.getDatasetResolver();
    Column column = resolver.getColumn(tableId, columnId);
    if (column == null) {
      violations.add(new SliceColumnNotFoundViolation(
          sliceReference(slice, context), slice.getId(), columnId, tableId.getObjectId()));
    }
  }

  private void validateTableReference(Slice slice, Identifier tableId,
      ValidationContext context, ImmutableList.Builder<Violation> violations) {
    DatasetResolver resolver = context.getDatasetResolver();
    Table table = resolver.getTable(tableId);
    if (table == null) {
      violations.add(new SliceTableNotFoundViolation(
          sliceReference(slice, context), slice.getId(), tableId.getObjectId()));
    }
  }

  private void validateId(Slice slice, ValidationContext context,
      ImmutableList.Builder<Violation> violations) {
    ItemReference sliceReference = sliceReference(slice, context);
    if (Strings.isNullOrEmpty(slice.getId())) {
      violations.add(new SliceNoIdViolation(sliceReference));
    } else {
      IdValidator validator = new IdValidator(sliceReference);
      validator.validate(slice.getId(), violations);
    }
  }

  private ItemReference sliceReference(Slice slice, ValidationContext context) {
    return ItemReference.reference(context.getCurrentDataset(), slice);
  }
}
